package com.ithuameng.notice.listener;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.ithuameng.notice.config.ApplicationContextProvider;
import com.ithuameng.notice.util.R;
import com.ithuameng.notice.netty.MyWebSocketHandler;
import com.rabbitmq.client.Channel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.data.redis.core.RedisTemplate;

import java.io.IOException;
import java.util.HashMap;

/**
 * 配置mq监听器
 */
@Slf4j
public class SysNoticeListener implements ChannelAwareMessageListener {

    private static ObjectMapper MAPPER = new ObjectMapper();

    private RedisTemplate redisTemplate = (RedisTemplate) ApplicationContextProvider.getApplicationContext().getBean("redisTemplate");

    @Override
    public void onMessage(Message message, Channel channel) throws IOException {
        // 判断此消息是否已经消费过
        String msgId = (String) message.getMessageProperties().getHeaders().get("spring_returned_message_correlation");
        long tag = message.getMessageProperties().getDeliveryTag();
        if (redisTemplate.opsForHash().entries("message_log").containsKey(msgId)) {
            log.info(msgId + ":消息已经被消费过");
            channel.basicAck(tag, false);
            return;
        }
        io.netty.channel.Channel wsChannel = MyWebSocketHandler.userChannelMap.get("admin");
        // 检查超级管理员是否在线
        if (wsChannel != null) {
            HashMap<String, Integer> countMap = new HashMap<>();
            countMap.put("sysNoticeCount", 1);
            R result = new R<>(true, 200, "查询成功", countMap);
            try {
                wsChannel.writeAndFlush(new TextWebSocketFrame(MAPPER.writeValueAsString(result)));
            } catch (Exception e) {
                // 消息处理失败 true阻止消息重新回到队列中两次消费
                log.info("netty channel writeAndFlush error...");
                channel.basicNack(tag, false, true);
            }
        }
        // 标记消息被处理过
        redisTemplate.opsForHash().put("message_log", msgId, "value");
        channel.basicAck(tag, false);
    }
}